package WindowAndWatermark

import Source.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.{AssignerWithPeriodicWatermarks, AssignerWithPunctuatedWatermarks}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.watermark.Watermark

object WatermarkCustom {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(1)
    //事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //设置周期
    env.getConfig.setAutoWatermarkInterval(150l)

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
      //      .assignTimestampsAndWatermarks(
      //      new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.milliseconds(3)) {
      //        override def extractTimestamp(t: SensorReading) = t.timeStamp
      //      })
      .assignTimestampsAndWatermarks(new PeriodicAssigner)

    //每十五秒统计窗口内个传感器的温度最小值,以及最新的时间戳
    val resultStream = dataStream
      .map(data => (data.id, data.temperature, data.timeStamp))
      .keyBy(_._1) //按照二元组第一个元素id分组
      .timeWindow(Time.seconds(5))
      .allowedLateness(Time.minutes(1)) //设置迟到数据
      .sideOutputLateData(new OutputTag[(String, Double, Long)]("late")) //迟到的流
      .reduce((CurRes, newData) => (CurRes._1, CurRes._2.min(newData._2), newData._3))

    resultStream.print("res")

    env.execute()
  }
}

/**
 * 自定义一个周期性的时间戳抽取
 */
class PeriodicAssigner extends AssignerWithPeriodicWatermarks[SensorReading] {
  val bound = 60 * 1000l //延时为1分钟

  var maxTS = Long.MinValue //观察到最大的时间戳

  override def getCurrentWatermark: Watermark = {
    new Watermark(maxTS - bound) //判断是否关闭窗口
  }

  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    maxTS = maxTS.max(t.timeStamp) //更新最大值
    t.timeStamp //取出时间戳
  }
}

/**
 * 自定义一个间断式生成watermark
 */
class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[SensorReading] {
  val bound = 60 * 1000l

  override def checkAndGetNextWatermark(t: SensorReading, l: Long): Watermark = {
    if (t.id == "sensor_1") {
      new Watermark(l - bound)
    } else {
      null
    }
  }

  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    t.timeStamp
  }
}